package test.org.offsync.async

import java.net.InetSocketAddress
import java.net.Socket
import java.net.SocketAddress
import java.util.concurrent.CyclicBarrier
import java.util.concurrent.atomic.AtomicInteger
import org.junit.runner.RunWith
import org.offsync.async.AsyncResult
import org.offsync.async.AsyncService
import org.offsync.async.AsyncServerSocket
import org.offsync.async.AsyncReturn
import org.offsync.util.Disposer
import org.offsync.util.Tasks
import org.offsync.util.Time
import org.specs._
import org.specs.runner._
import org.offsync.util.DisposeState
import scala.collection.{mutable => mutable}
import org.offsync.async.AsyncSocketChannel

@RunWith(classOf[JUnitSuiteRunner])
class SpecAsyncService extends SpecificationWithJUnit("test.org.offsync.util.SpecAsyncService") {
  "New AsyncService object can dispose" in {
    val asyncService = new AsyncService()
    asyncService.waitDisposed(Time.now)
    asyncService.disposeState must_== DisposeState.Live
  }

  "AsyncService object can waitDisposed" in {
    val asyncService = new AsyncService()
    var counter = 0
    val thread = new Thread() {
      override def run(): Unit = {
        Thread.sleep(10)
        counter *= 2
        asyncService.dispose()
      }
    }
    thread.start()
    asyncService.waitDisposed()
    counter += 1
    thread.join()
    asyncService.disposeState must_== DisposeState.Disposed
  }
  
  "AsyncService object can dispose when accepting in AsyncServerSocket object" in {
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        disposer.disposes(socketChannel)
        barrier.await()
        AsyncReturn
      }
      val socket = new Socket()
      
      try {
        socket.connect(new InetSocketAddress("localhost", 9876))
        Thread.sleep(100)
      } finally {
        socket.close()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
  }
  
  "AsyncService object can dispose when reading in AsyncServerSocket object" in  {
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        socketChannel.read(asyncService, 1) { result =>
          val maybeArray = result.value
          maybeArray must_== None
          barrier.await()
          AsyncReturn
        }
      }
      val socket = new Socket()
      
      try {
        socket.connect(new InetSocketAddress("localhost", 9876))
        
        Thread.sleep(100)
      } finally {
        socket.close()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
  }

  "AsyncService can read" in  {
    var assertions = new Tasks
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        socketChannel.read(asyncService, 1) { result =>
          assertions.add { result.value.toList(0).toList must_== List[Byte](42) }
          socketChannel.read(asyncService, 1) { result =>
            assertions.add { result.value.toList(0).toList must_== List[Byte](43) }
            barrier.await()
            AsyncReturn
          }
        }
      }
      val socket = new Socket()
      
      try {
        socket.connect(new InetSocketAddress("localhost", 9876))
        Thread.sleep(100)
        socket.getOutputStream().write(Array[Byte](42))
        Thread.sleep(100)
        socket.getOutputStream().write(Array[Byte](43))
        
        Thread.sleep(100)
      } finally {
        socket.close()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
    assertions.run()
  }

  "AsyncService can read and write" in  {
    var assertions = new Tasks
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        socketChannel.read(asyncService, 1) { result =>
          assertions.add(result.value.toList(0).toList must_== List[Byte](42))
          socketChannel.read(asyncService, 1) { result =>
            assertions.add(result.value.toList(0).toList must_== List[Byte](44))
            barrier.await()
            AsyncReturn
          }
        }
      }
      
      val writeSocketChannel = AsyncSocketChannel.open(new InetSocketAddress("localhost", 9876))
      
      try {
        writeSocketChannel.write(asyncService, Array[Byte](42)) { result =>
          writeSocketChannel.write(asyncService, Array[Byte](44)) { result =>
            AsyncReturn
          }
        }
        
        Thread.sleep(100)
      } finally {
        writeSocketChannel.dispose()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
    assertions.run()
  }
}
